package net.techfuser.storesalereport;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.*;
import java.text.SimpleDateFormat;

class OrderMySqlSink extends RichSinkFunction<Order> {

    private Connection connection = null;
    private PreparedStatement preparedStatement = null;

    @Override
    public void open(Configuration parameters) throws Exception {

        String url = "jdbc:mysql://qnh-flink-data.mysql.rds.aliyuncs.com:3306/flink_data_output?useUnicode=true&characterEncoding=utf-8";
        String user = "qnh_flink";
        String passwd = "Md7fpbWSHpBa";
//        String url = "jdbc:mysql://127.0.0.1:3306/qnh?useUnicode=true&characterEncoding=utf-8";
//        String user = "root";
//        String passwd = "";
//        Class.forName("com.mysql.jdbc.Driver");
        connection = DriverManager.getConnection(url, user, passwd);
    }

    @Override
    public void invoke(Order order, Context context) throws Exception {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");

        preparedStatement = connection.prepareStatement(
            "insert into store_sale_summary" +
                    "(ent_id, region_code, region_name, channel_keyword, confirm_time, " +
                    "valid_order_num, total_original_value, total_sale_value, " +
                    "total_disc_value, total_sj_disc, total_pt_disc," +
                    "total_dp_disc, total_sj_dp, total_pt_dp, logistics_value, package_fee) " +
                    "values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) ON DUPLICATE KEY UPDATE " +

                "valid_order_num = valid_order_num + ?," +
                "total_original_value = total_original_value + ?," +
                "total_sale_value = total_sale_value + ?," +
                "total_disc_value = total_disc_value + ?," +
                "total_sj_disc = total_sj_disc + ?," +
                "total_pt_disc = total_pt_disc + ?," +
                "total_dp_disc = total_dp_disc + ?," +
                "total_sj_dp = total_sj_dp + ?," +
                "total_pt_dp = total_pt_dp + ?," +
                "logistics_value = logistics_value + ?," +
                "package_fee = package_fee + ?," +
                "last_modify_time =  ? " +
                "where ent_id = ? and region_code = ? and channel_keyword = ? and confirm_time = ?");
        // insert 部分参数
        preparedStatement.setLong(1, order.getEnt_id());
        preparedStatement.setString(2, order.getRegion_code());
        preparedStatement.setString(3, order.getRegion_name());
        preparedStatement.setString(4, order.getChannel_keyword());
        preparedStatement.setDate(5, new Date(order.getConfirm_time().getTime()));

        preparedStatement.setInt(6, order.getValid_order_num());
        preparedStatement.setDouble(7, order.getTotal_original_value());
        preparedStatement.setDouble(8, order.getTotal_sale_value());
        preparedStatement.setDouble(9, order.getTotal_disc_value());
        preparedStatement.setDouble(10, order.getTotal_sj_disc());
        preparedStatement.setDouble(11, order.getTotal_pt_disc());
        preparedStatement.setDouble(12, order.getTotal_dp_disc());
        preparedStatement.setDouble(13, order.getTotal_sj_dp());
        preparedStatement.setDouble(14, order.getTotal_pt_dp());
        preparedStatement.setDouble(15, order.getLogistics_value());
        preparedStatement.setDouble(16, order.getPackage_fee());


        // update 部分参数
        preparedStatement.setInt(17, order.getValid_order_num());
        preparedStatement.setDouble(18, order.getTotal_original_value());
        preparedStatement.setDouble(19, order.getTotal_sale_value());
        preparedStatement.setDouble(20, order.getTotal_disc_value());
        preparedStatement.setDouble(21, order.getTotal_sj_disc());
        preparedStatement.setDouble(22, order.getTotal_pt_disc());
        preparedStatement.setDouble(23, order.getTotal_dp_disc());
        preparedStatement.setDouble(24, order.getTotal_sj_dp());
        preparedStatement.setDouble(25, order.getTotal_pt_dp());
        preparedStatement.setDouble(26, order.getLogistics_value());
        preparedStatement.setDouble(27, order.getPackage_fee());
        preparedStatement.setTimestamp(28, new Timestamp(System.currentTimeMillis()));
        // update where 部分参数
        preparedStatement.setLong(29, order.getEnt_id());
        preparedStatement.setString(30, order.getRegion_code());
        preparedStatement.setString(31, order.getChannel_keyword());
        preparedStatement.setString(32, sdf.format(order.getConfirm_time()));

        preparedStatement.execute();
    }

    @Override
    public void close() throws Exception {
        if (null != connection) connection.close();
        if (null != preparedStatement) preparedStatement.close();
    }

}